/**
 * 
 */
package io.dora.registry.curator;

import io.dora.registry.zk.ZkUtils;

import java.util.Arrays;
import java.util.HashSet;
import java.util.Set;

import org.apache.commons.lang.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Curator ZK 工具
 *
 * @author zhaohongbo
 *
 */
public class CuratorUtils {

	private static Logger LOGGER = LoggerFactory.getLogger(CuratorRegistry.class);

	/**
	 * 创建一个zk node
	 *
	 * @param zkClient zk客户端
	 * @param path     数据路径
	 * @return
	 */
	public static boolean createNode(CuratorFramework zkClient, String path) {
		boolean flag = true;
		try {
			Stat stat = zkClient.checkExists().forPath(path);
			if (stat == null) {
				zkClient.create().
						creatingParentsIfNeeded().
						withMode(CreateMode.PERSISTENT).
						withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE).
						forPath(path);
			}
		} catch (Exception e) {
			flag = false;
			LOGGER.info("{}", e);
		}
		return flag;
	}
	
	/**
	 * 添加或更新node数据
	 *
	 * @param zkClient zk客户端
	 * @param path     数据路径
	 * @param data     节点数据
	 * @param append   是否追加
	 */
	public static void addOrUpdateNodeData(CuratorFramework zkClient, String path, String data, boolean append) {
		try {
			
			if (append) {
				String existsData = getNodeData(zkClient, path);
				if (StringUtils.isNotBlank(existsData)) {
					StringBuffer dataBuffer = new StringBuffer(existsData);
					dataBuffer.append(ZkUtils.spliter).append(data);
					data = dataBuffer.toString();
					String[] arrays = StringUtils.split(data, ZkUtils.spliter);
					Set<String> dataSet = new HashSet<String>(Arrays.asList(arrays));
					data = StringUtils.join(dataSet, ZkUtils.spliter);
				}
			}
			
			LOGGER.info("path {} data  {}", path, data);
			
			Stat stat = zkClient.checkExists().forPath(path);
			if (stat != null) {
				zkClient.setData().forPath(path, data.getBytes());
			}
		} catch (Exception e) {
			LOGGER.info("{}", e);
		}
	}

	/**
	 * 获取节点数据
	 *
	 * @param zkClient  zk客户端
	 * @param path      数据路径
	 * @return
	 */
	public static String getNodeData(CuratorFramework zkClient, String path) {
		String data = "";
		try {
			byte[] bdata = zkClient.getData().forPath(path);
			if (bdata != null) {
				data = new String(bdata);
			}
		} catch (Exception e) {
			LOGGER.info("{}", e);
		}
		return data;
	}

	/**
	 * 删除节点数据
	 *
	 * @param zkClient    zk客户端
	 * @param path        数据路径
	 * @param removeData  被删除的数据
	 */
	public static void removeData(CuratorFramework zkClient, String path, String removeData) {
		try {
			String existsData = getNodeData(zkClient, path);
			if (StringUtils.isNotBlank(existsData)) {
				String[] arrays = StringUtils.split(existsData, ZkUtils.spliter);
				Set<String> dataSet = new HashSet<String>(Arrays.asList(arrays));
				dataSet.remove(removeData);
				String data = StringUtils.join(dataSet, ZkUtils.spliter);
				addOrUpdateNodeData(zkClient, path, data, false);
			}
		} catch (Exception e) {
			LOGGER.info("{}", e);
		}
	}
}
